package com.gagakuai.infrastructure.utils;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;
import java.io.InputStream;

/**
 * @description: 参数工具类
 * @author: houhong
 * @create: 2024-09-07 20:00
 **/
public class ParameterUtils {

    /**
     * 默认配置文件
     */
    private static final String DEFAULT_CONFIG = ParameterConstantsUtil.FLINK_ROOT_FILE;
    /**
     * 带环境配置文件
     */
    private static final String FLINK_ENV_FILE = ParameterConstantsUtil.FLINK_ENV_FILE;
    /**
     * 环境变量
     */
    private static final String ENV_ACTIVE = ParameterConstantsUtil.FLINK_ENV_ACTIVE;

    public static ParameterTool getParameters() {


        InputStream inputStream = ParameterUtils.class.getClassLoader().getResourceAsStream(DEFAULT_CONFIG);


        try {
            ParameterTool defaultPropertiesFile =
                    ParameterTool.fromPropertiesFile(inputStream);

            //获取环境参数
            String envActive = getEnvActiveValue(defaultPropertiesFile);

            //读取真正的配置环境 (推荐使用 Thread.currentThread() 读取配置文件)
            return ParameterTool
                    // ParameterTool读取变量优先级 系统环境变量>启动参数变量>配置文件变量
                    // 从配置文件获取配置
                    .fromPropertiesFile(
                            //当前线程
                            Thread.currentThread()
                                    //返回该线程的上下文信息, 获取类加载器
                                    .getContextClassLoader()
                                    .getResourceAsStream(envActive))
                    // 从系统环境变量获取配置
                    .mergeWith(ParameterTool.fromSystemProperties());
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    public static ParameterTool getParameters(final String[] args) {

        InputStream inputStream = ParameterUtils.class.getClassLoader().getResourceAsStream(DEFAULT_CONFIG);

        try {

            ParameterTool defaultPropertiesFile = ParameterTool.fromPropertiesFile(inputStream);

            String envActive = getEnvActiveValue(defaultPropertiesFile);

            //读取真正的配置环境 (推荐使用 Thread.currentThread() 读取配置文件)
            return ParameterTool
                    // ParameterTool读取变量优先级 系统环境变量>启动参数变量>配置文件变量
                    // 从配置文件获取配置
                    .fromPropertiesFile(
                            //当前线程
                            Thread.currentThread()
                                    //返回该线程的上下文信息, 获取类加载器
                                    .getContextClassLoader()
                                    .getResourceAsStream(envActive))
                    // 从启动参数中获取配置
                    .mergeWith(ParameterTool.fromArgs(args))
                    // 从系统环境变量获取配置
                    .mergeWith(ParameterTool.fromSystemProperties());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }


    private static String getEnvActiveValue(ParameterTool defaultPropertiesFile) {
        //选择参数环境
        String envActive = null;
        if (defaultPropertiesFile.has(ENV_ACTIVE)) {
            envActive = String.format(FLINK_ENV_FILE, defaultPropertiesFile.get(ENV_ACTIVE));
        }

        return envActive;
    }


    public static void envWithConfig(StreamExecutionEnvironment env, ParameterTool parameterTool) {

        //##########  checkpoint 设置 ############//
        //每60秒触发checkpoint
        env.enableCheckpointing(parameterTool.getInt(ParameterConstantsUtil.FLINK_CHECKPOINT_INTERVAL));
        CheckpointConfig ck = env.getCheckpointConfig();
        // checkpoint 必须在60秒内结束，否则被丢弃
        ck.setCheckpointTimeout(parameterTool.getInt(ParameterConstantsUtil.FLINK_CHECKPOINT_TIMEOUT));
        //checkpoint间最小间隔 30秒 (指定了这个值, setMaxConcurrentCheckpoints自动默认为1)
        ck.setMinPauseBetweenCheckpoints(parameterTool.getInt(ParameterConstantsUtil.FLINK_CHECKPOINT_MINPAUSE));
        // checkpoint 语义设置为 精确一致( EXACTLY_ONCE )
        ck.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 最多允许 checkpoint 失败 3 次
        ck.setTolerableCheckpointFailureNumber(parameterTool.getInt(ParameterConstantsUtil.FLINK_CHECKPOINT_FAILURENUMBER));
        // 同一时间只允许一个 checkpoint 进行
        ck.setMaxConcurrentCheckpoints(parameterTool.getInt(ParameterConstantsUtil.FLINK_CHECKPOINT_MAXCONCURRENT));

        //##########  checkpoint 设置 end ############//
        //设置 State 存储
        env.setStateBackend(new HashMapStateBackend());
        //并行度设置
        env.setParallelism(parameterTool.getInt(ParameterConstantsUtil.FLINK_PARALLELISM));

    }
}
